/**
 * 
 */
package name.yzhu.wbdata.io;

import java.io.IOException;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import name.yzhu.cloud.CloudException;
import name.yzhu.cloud.hbase.SingleTableWriter;
import name.yzhu.cloud.hbase.YamlHBaseRowBase;
import name.yzhu.common.conf.AbstractConfig;
import name.yzhu.common.service.Stoppable;
import name.yzhu.common.util.IdentifyUnique;
import name.yzhu.common.yaml.YamlItem;
import name.yzhu.wbdata.Constant;

/**
 * @author yhzhu
 *
 */
public class PersistHBase implements IPersistWeibo {
	private static Logger LOG = Logger.getLogger(PersistHBase.class);
	private static final String HBASE_POOL_SIZE = "hbase.pool.size";
	
	private AbstractConfig cfg = null;
	private SingleTableWriter writer = null;
	private Stoppable stop = null;
	private Configuration hbaseCFG = null;

	public PersistHBase(AbstractConfig cfg, Configuration hbaseCFG, Stoppable stop) {
		this.cfg = cfg;
		this.hbaseCFG  = hbaseCFG;
		this.stop = stop;
	}
	
	/* (non-Javadoc)
	 * @see name.yzhu.wbdata.io.IPersistWeibo#persistYaml(java.util.List)
	 */
	@Override
	public void persistYaml(List<YamlItem> yamls) throws IOException {
		for (YamlItem yaml : yamls)
			persistYaml(yaml);
	}

	/* (non-Javadoc)
	 * @see name.yzhu.wbdata.io.IPersistWeibo#persistYaml(name.yzhu.common.yaml.YamlItem)
	 */
	@Override
	public void persistYaml(YamlItem yaml) throws IOException {
		YamlHBaseRowBase row = new WeiboYamlRow(yaml);
		String key = (String) yaml.remove(name.yzhu.common.Constant.YAML_FIELD_UUID);
		row.setKey(Bytes.toBytes(key));
		try {
			writer.write(row);
		} catch (CloudException e) {
			LOG.error(e.getMessage(), e);
		}
	}

	/* (non-Javadoc)
	 * @see name.yzhu.wbdata.io.IPersistWeibo#open()
	 */
	@Override
	public void open() throws IOException {
		
		int size = Runtime.getRuntime().availableProcessors();
		size = cfg.getInt(HBASE_POOL_SIZE, size);
		writer  = new SingleTableWriter(hbaseCFG, Constant.HBASE_WEIBO_TABLE, size, stop );
	}

	/* (non-Javadoc)
	 * @see name.yzhu.wbdata.io.IPersistWeibo#close()
	 */
	@Override
	public void close() throws IOException {
		writer.close();
	}

	/* (non-Javadoc)
	 * @see name.yzhu.wbdata.io.IPersistWeibo#setIdentifyUniqueue(name.yzhu.common.util.IdentifyUnique)
	 */
	@Override
	public void setIdentifyUniqueue(IdentifyUnique identify) {
	}

}
